package com.fulihui.wgj.schedule.job;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.fulihui.common.constants.TopicConstants;
import com.fulihui.common.request.rpc.AppRequest;
import com.fulihui.common.util.JedisUtil;
import com.fulihui.common.util.TenMinutesMsgUtil;
import com.fulihui.starter.kafka.sdk.KafkaProducerSdk;
import com.fulihui.wgj.facade.dto.Wx48NewsInfoDTO;
import com.fulihui.wgj.facade.msg.Wx48HMsg;
import com.fulihui.wgj.facade.service.wx48.Wx48NewsInfoService;
import com.fulihui.wgj.schedule.annotation.MutexLock;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @author: LiuJie
 * @version: 2018/7/5 17:00
 * @description: 十分钟刷新
 */
@Component
@Slf4j
public class TenMinutesSendMsgJob {

    @Autowired
    TenMinutesMsgUtil tenMinutesMsgUtil;
    @Autowired
    JedisUtil jedisUtil;
    @Autowired
    KafkaProducerSdk kafkaProducerSdk;
    @Reference(version = "1.0.0")
    Wx48NewsInfoService wx48NewsInfoService;

    @MutexLock(node = "${job.node.tenMinutesSendMsgJob_execute}")
    @Scheduled(cron = "${job.cron.tenMinutesSendMsgJob_execute}")
    public void execute(){
        log.info("十分钟用户扫描。开始");
        specificUsers48HProduce();
        log.info("十分钟用户扫描。结束");
    }

    private void specificUsers48HProduce() {
        Date anHourAgo = DateUtils.addHours(new Date(), -1);
        Date tenMinutesAgo = DateUtils.addMinutes(new Date(), -10);
        // 不同公众号不同的key
        Set<String> keys = tenMinutesMsgUtil.keys();
        log.info("十分钟消息扫描到的 keys={}", keys);
        keys.forEach(key -> {
            Set<String> openIds = jedisUtil.zrangeByScore(key, anHourAgo.getTime(), tenMinutesAgo.getTime());
            if (CollectionUtils.isEmpty(openIds)) {
                log.info("十分钟用户消息 key = {} 中不存在用户",key);
                return;
            }
            String[] splitKey = StringUtils.split(key, TenMinutesMsgUtil.REDIS_SUBSCRIBE_USER_KEY);
            if (splitKey != null && splitKey.length > 0) {
                String appId = splitKey[0];

                // 推送数据
                if(putMsg2Mq(appId, new ArrayList<>(openIds))){
                    // 删除发送过的缓存用户
                    jedisUtil.zrem(key,openIds.toArray(new String[openIds.size()]));
                }

            }
        });
    }

    /**
     * 推送到 kafka
     *
     * @param appId
     * @param openIds
     */
    private boolean putMsg2Mq(String appId, List<String> openIds) {

        String key = TenMinutesMsgUtil.TEN_MINUTES_NEWS_KEY+appId;
        Wx48HMsg weixin48HMsg = new Wx48HMsg();
        weixin48HMsg.setAppId(appId);
        weixin48HMsg.setNewsKey(key);

        AppRequest appRequest = new AppRequest();
        appRequest.setAppId(appId);
        // 把要发送的消息模板写入 redis
        Wx48NewsInfoDTO data = wx48NewsInfoService.querySpecificReply(appRequest).getData();
        if(null == data){
            log.error("十分钟消息推送失败！！获取特定模板数据为空,appid={}",appId);
            return true;//依旧删除 redis
        }
        jedisUtil.set(key,data.getXmlContent());

        //消息写入kafka
        List<String> newOpenIds = Lists.newArrayList();
        int i = 0;
        for (String openId : openIds) {
            // 每100条触发一次生产
            if ( ++i % TenMinutesMsgUtil.mqLengthGenerally == 0) {
                weixin48HMsg.setOpenIds(newOpenIds);
                kafkaProducerSdk.send(TopicConstants.WX_48H_MSG, JSON.toJSONString(weixin48HMsg));
                newOpenIds.clear();
            }
            newOpenIds.add(openId);
        }
        // 最后不到100条的数据
        weixin48HMsg.setOpenIds(newOpenIds);
        weixin48HMsg.setLast(true);
        kafkaProducerSdk.send(TopicConstants.WX_48H_MSG, JSON.toJSONString(weixin48HMsg));
        return true;
    }
}
